package com.zlm.app;

import com.zlm.bean.OrderEvent;
import com.zlm.bean.ReceiptEvent;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoProcessFunction;
import org.apache.flink.streaming.api.functions.timestamps.AscendingTimestampExtractor;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.net.URL;

/**
 * Author: Harbour
 * Date: 2021-05-30 21:22
 * Desc: 匹配订单流与支付流
 */
public class TxPayMatchAPP {
    public static void main(String[] args) throws Exception {
        // step 1 获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        // step 2 获取数据源
        URL resource = TxPayMatchAPP.class.getResource("/OrderLog.csv");
        DataStream<String> orderInputDataStream = env.readTextFile(resource.getPath());

        URL receiptResource = TxPayMatchAPP.class.getResource("/ReceiptLog.csv");
        DataStream<String> receiptInputDataStream = env.readTextFile(receiptResource.getPath());

        // step 3 转换数据类型
        DataStream<OrderEvent> orderEventDataStream = orderInputDataStream
                .map(new MapFunction<String, OrderEvent>() {
                    @Override
                    public OrderEvent map(String value) throws Exception {
                        String[] fields = value.split(",");

                        return new OrderEvent(
                                Long.parseLong(fields[0]),
                                fields[1],
                                fields[2],
                                Long.parseLong(fields[3])
                        );
                    }
                })
                .assignTimestampsAndWatermarks(new AscendingTimestampExtractor<OrderEvent>() {
                    @Override
                    public long extractAscendingTimestamp(OrderEvent element) {
                        return element.getTimestamp() * 1000L;
                    }
                });

        DataStream<ReceiptEvent> receiptEventDataStream = receiptInputDataStream
                .map(new MapFunction<String, ReceiptEvent>() {
                    @Override
                    public ReceiptEvent map(String value) throws Exception {
                        String[] fields = value.split(",");

                        return new ReceiptEvent(
                                fields[0],
                                fields[1],
                                Long.parseLong(fields[2])
                        );
                    }
                })
                .assignTimestampsAndWatermarks(new AscendingTimestampExtractor<ReceiptEvent>() {
                    @Override
                    public long extractAscendingTimestamp(ReceiptEvent element) {
                        return element.getTimestamp() * 1000L;
                    }
                });


        // step 4 合并订单流与支付流
        SingleOutputStreamOperator<Tuple2<OrderEvent, ReceiptEvent>> connectDataStream = orderEventDataStream.keyBy(OrderEvent::getTxId)
                .connect(receiptEventDataStream.keyBy(ReceiptEvent::getTxId))
                .process(new MyCoProcessFunction());


        // step 5 输出执行结果与侧输出流
//        connectDataStream.print();
        connectDataStream.getSideOutput(orderEventNoMatchTag).print();
        connectDataStream.getSideOutput(receiptEventNoMatchTag).print();

        env.execute();
    }

    private final static OutputTag<OrderEvent> orderEventNoMatchTag = new OutputTag<OrderEvent>("orderEventNoMatch") {};
    private final static OutputTag<ReceiptEvent> receiptEventNoMatchTag = new OutputTag<ReceiptEvent>("receiptEventNoMatch") {};

    private static class MyCoProcessFunction extends CoProcessFunction<OrderEvent, ReceiptEvent, Tuple2<OrderEvent, ReceiptEvent>> {

        private ValueState<OrderEvent> orderState;
        private ValueState<ReceiptEvent> receiptState;

        @Override
        public void open(Configuration parameters) throws Exception {
            orderState = getRuntimeContext().getState(new ValueStateDescriptor<OrderEvent>("orderEvent", OrderEvent.class));
            receiptState = getRuntimeContext().getState(new ValueStateDescriptor<ReceiptEvent>("receiptEvent", ReceiptEvent.class));
        }

        @Override
        public void processElement1(OrderEvent orderEvent, Context ctx, Collector<Tuple2<OrderEvent, ReceiptEvent>> out) throws Exception {
            ReceiptEvent receiptEvent = receiptState.value();
            if (receiptEvent != null) {
                out.collect(new Tuple2<OrderEvent, ReceiptEvent>(orderEvent, receiptEvent));
                receiptState.clear();
            } else {
                orderState.update(orderEvent);
                ctx.timerService().registerEventTimeTimer(orderEvent.getTimestamp() * 1000L + 5000L);
            }
        }

        @Override
        public void processElement2(ReceiptEvent receiptEvent, Context ctx, Collector<Tuple2<OrderEvent, ReceiptEvent>> out) throws Exception {
            OrderEvent orderEvent = orderState.value();
            if (orderEvent != null) {
                out.collect(new Tuple2<OrderEvent, ReceiptEvent>(orderEvent, receiptEvent));
                orderState.clear();
            } else {
                receiptState.update(receiptEvent);
                ctx.timerService().registerEventTimeTimer(receiptEvent.getTimestamp() * 1000L + 3000L);
            }
        }

        @Override
        public void onTimer(long timestamp, OnTimerContext ctx, Collector<Tuple2<OrderEvent, ReceiptEvent>> out) throws Exception {
            if (orderState.value() != null) {
                ctx.output(
                    orderEventNoMatchTag,
                    orderState.value()
                );
            }

            if (receiptState.value() != null) {
                ctx.output(
                    receiptEventNoMatchTag,
                    receiptState.value()
                );
            }

            receiptState.clear();
            orderState.clear();
        }
    }
}
